package com.itcast.flink.usage.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static java.util.regex.Pattern.compile;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
public class BatchProcessorApplication {
    
    public static void main(String[] args) throws Exception {
        // 1. 定义Flink运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 读取数据源 （日志文件信息）
        DataSource<String> logData = env.readTextFile("./data/order_info.log");
        
        // 3. 清洗转换数据
        logData.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                // 1) 根据正则， 提取每行日志的级别
                Pattern pattern = compile("\\[main\\](.*?)\\[");
                Matcher matcher = pattern.matcher(value);
                if (matcher.find()) {
                    // 2) 如果匹配符合规则， 放置元组， 输出数据
                    collector.collect(new Tuple2<String, Integer>(matcher.group(1).trim(), 1));
                }
            }
        }).groupBy(0).sum(1).print(); // 4. 根据日志级别， 汇总统计日志级别的次数，打印结果
    }
}
